#include "rpcclient.h"

#include <google/protobuf/descriptor.h>
#include <google/protobuf/service.h>
#include "../proto/rpcheader.pb.h"
#include "../proto/rpcresponse.pb.h"
#include "tinyrpc/base/stringUtil.h"
#include "tinyrpc/net/inetaddress.h"
#include "rpcService.h"
#include "tinyrpc/net/coroutine_sync.h"

#include <netinet/in.h>
#include <sys/types.h>          /* See NOTES */
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <unistd.h>

namespace TinyRPC{

// 所有通过stub 代理对象调用rpc的方法， 都走到这里了，统一做rpc方法调用的数据序列化
void TinyRPCClient::CallMethod(const google::protobuf::MethodDescriptor* method,
                    google::protobuf::RpcController* controller, const google::protobuf::Message* request,
                    google::protobuf::Message* response, google::protobuf::Closure* done)
{
    LOG_DEBUG<<"TinyRPCClient:: CallMethod";
    const google::protobuf::ServiceDescriptor * sd = method->service();
    std::string service_name = sd->name();
    std::string method_name = method->name();

    std::string args_str;
    request->SerializeToString(&args_str);

    RPCHeader rpcHeader;
    rpcHeader.set_service_name(service_name);
    rpcHeader.set_method_name(method_name);
    rpcHeader.set_args(args_str);
    RPCRequestInfoPtr info = *(RPCRequestInfoPtr*)done;
    int msg_no = info->getNo();
    rpcHeader.set_msg_no(msg_no);

    int header_size = 0;
    std::string rpc_header_str;
    rpcHeader.SerializeToString(&rpc_header_str);
    header_size = rpc_header_str.size();

    // std::string send_rpc_str;
    // send_rpc_str.insert(0, std::string((char*)&header_size, sizeof(int)));
    // send_rpc_str += rpc_header_str;

    // 
    RPCServicePtr rpcService = getService(service_name);
    if(rpcService == nullptr){
        // 失败了该如何处理
        EventLoop::CurrentEventLoop()->queueInLoop(Task(std::bind(&TinyRPCClient::co_updateServiceInfo, this), true));
        info->setFail();
    }else{
        try{
            LOG_DEBUG<<"send to service " <<rpc_header_str <<" " <<msg_no <<" info trytime" <<info->getTryTimes();
            std::weak_ptr<RPCRequestInfo> weak_info(info);
            rpcService->getSockSocketChannel()->write(rpc_header_str, m_codec);
            
            getNextLoop()->runAfter(0.5, Task([=](){
                RPCRequestInfoPtr info2 = weak_info.lock();
                if(info2 && info2->getStatus() == SEND){
                    info2->setTimeOut();
                }
                
            }, false));
        }catch(...){
            getNextLoop()->queueInLoop(Task(std::bind(&TinyRPCClient::co_updateServiceInfo, this), true));
            info->setFail();
        }
        
    }
    
    
}

RPCServicePtr TinyRPCClient::getService(std::string service_name){
    auto it = m_serviceMap.find(service_name);
    std::unordered_map<std::string, RPCServicePtr> & services = it->second->m_services;
    
    LockGuard<CoLock> gurad(it->second->m_lock);
    auto& idx = it->second->idx;
    if(idx == services.end()){
        idx = services.begin();
    }
    
    while(idx->second->getStatus() != DEAD){
        // 每一次建立一个新连接都应该 connect 得到一个SocketChannel
         if(idx->second->getStatus() == RPCServiceINIT || idx->second->getStatus() == DISCONNECTED ){
            idx->second->connect();

            if(idx->second->getStatus() == DEAD){
                continue;
            }else{
                LOG_INFO<<"create new receiveResponseFromServer coroutine";
                CoroutinePtr co = std::make_shared<Coroutine>(std::bind(&TinyRPCClient::co_receiveResponseFromServer,this,idx->second));
                idx->second->setCoroutinePtr(co);
                co->setStatus(CoroutineStatus::READY);
                EventLoop::CurrentEventLoop()->queueInLoop(Task(co));
                // CoWakeUp(co);
                // EventLoop::CurrentEventLoop()->wakeup();
            }
        }
        RPCServicePtr servicePtr = idx->second;
        idx ++;
        if(idx == services.end()){
            idx = services.begin();
        }
        return servicePtr;
    }
    return nullptr;
}

void TinyRPCClient::co_receiveResponseFromServer(RPCServicePtr service){
    
    while(true){
        while( service->getStatus() == DISCONNECTED){
            LOG_INFO<<"co_receiveResponseFromServer yield";
            Coroutine::Yield();
        }
       
        int ret = service->getSockSocketChannel()->receiveFromRemote();

        if(ret == 0){
            LOG_INFO<<"co_receiveResponseFromServer close";
            service->closeConn();
            break;
        }

        std::string response_str = service->getSockSocketChannel()->read(m_codec);
        if(response_str.size() == 0) continue;

        RPCResponse rpcResponse;
        rpcResponse.ParseFromString(response_str);
        LOG_DEBUG<<response_str << " recvived "<<rpcResponse.msg_no();
        RPCRequestInfoPtr info = m_requestMessageManager.getRequestInfo(rpcResponse.msg_no());
        // 失败
        if(rpcResponse.code() == 0){
            // 失败了该如何处理
            info->setFail();
        }else{
            info->setSuccAndParse(rpcResponse.response());
        }
        //RPCService.GetService(service_name).send(send_rpc_str);
    }
}


void TinyRPCClient::co_updateServiceInfo(){
    LOG_INFO<<"co_updateServiceInfo begin";
    for(auto it : m_serviceMap){
        m_redisRegister.co_updateServiceInfo(it.first, it.second, true);
    }

}

void TinyRPCClient::registServiceStub(::google::protobuf::Service* service){
    const google::protobuf::ServiceDescriptor * serviceDescrip = service->GetDescriptor();
    std::string serviceName = serviceDescrip->name();

    int methodCount = serviceDescrip->method_count();
    RPCServiceInfoPtr serviceInfo = std::make_shared<RPCServiceInfo>();
    for(int i=0;i<methodCount; i++){
        
        serviceInfo->m_service = service;
        const google::protobuf::MethodDescriptor * methodDescrip = serviceDescrip->method(i);
        serviceInfo->m_methodMap.emplace(methodDescrip->name(), methodDescrip);  
        
    }
    m_serviceMap.insert({serviceName, serviceInfo});

    m_redisRegister.co_updateServiceInfo(serviceName, serviceInfo, false);
    //getNextLoop()->runInLoop(Task(std::bind(&TinyRPCClient::keyEventsubscribe, this), true));
    LOG_INFO<<"registServiceStub after";
}

void TinyRPCClient::co_discoverRPCService(){
    int sockfd = socket(AF_INET, SOCK_STREAM, 0);
    std::string redis_ip = m_config.LookUp<std::string>("redis.ip");
    int redis_port = m_config.LookUp<int>("redis.port");
    INetAddress redisAddress(redis_ip, redis_port);
    SocketChannelPtr subscribeSocketChannel = std::make_shared<SocketChannel>(sockfd, redisAddress);
    int ret = subscribeSocketChannel->connect(redisAddress);
    if(ret < 0){
        
        return ;
    }
    std::vector<std::string> redis_channels = {"ADDSERVICE"};
   
    m_redisRegister.co_subscribe(subscribeSocketChannel,redis_channels, Task([&](){
        getNextLoop()->queueInLoop(Task(std::bind(&TinyRPCClient::co_updateServiceInfo, this), true));
    }, false));
        
}

// 在requestmanager 里面删除掉
void TinyRPCClient::delRPCRequestInfo(RPCRequestInfoPtr infoPtr){
    m_requestMessageManager.deleteRequestInfoPtr(infoPtr->getNo());
}
}




